/**
 * 
 */
package com.scalyze.apps.socialanyltics.auth.impl;

import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLEncoder;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.scalyze.apps.socialanyltics.auth.api.Collector;
import com.scalyze.apps.socialanyltics.collector.Dequeuer;
import com.scalyze.apps.socialanyltics.datastore.JSONService;

/**
 * @author vivek.mishra
 * 
 */
public class TwitterCollector implements Collector
{
    public static final Collector INSTANCE = new TwitterCollector();

    private static final Log logger = LogFactory.getLog(TwitterCollector.class);

    private HttpURLConnection connection;

    private LinkedBlockingQueue<String> messageQueue = new LinkedBlockingQueue<String>();
    
    private JSONService service = new JSONService();

    /*
     * (non-Javadoc)
     * 
     * @see
     * com.scalyze.apps.socialanyltics.auth.api.Authenticator#authenticate(java
     * .lang.String, java.lang.String)
     */
    @Override
    public void authenticate(String params, String connectionUrl)
    {

        String encodedParams = null;
        try
        {
            encodedParams = URLEncoder.encode(params, "UTF-8");
            encodedParams = encodedParams.replace("+", "%20");
            String urlStr = connectionUrl + "?track=" + encodedParams;
            URL url = new URL(urlStr);
            connection = (HttpURLConnection) url.openConnection(/* prox */);
        }
        catch (UnsupportedEncodingException e)
        {
            logger.error(e.getMessage());
            throw new AuthenticationException(e.getMessage());
        }
        catch (MalformedURLException e)
        {
            logger.error(e.getMessage());
            throw new AuthenticationException(e.getMessage());
        }
        catch (IOException e)
        {
            throw new AuthenticationException(e.getMessage());
        }

    }

    /*
     * (non-Javadoc)
     * 
     * @see com.scalyze.apps.socialanyltics.auth.api.Authenticator#getStream()
     */
    @Override
    public InputStream getStream()
    {
        validate();
        try
        {
            return connection.getInputStream();
        }
        catch (IOException e)
        {
            logger.error(e.getMessage());
            throw new AuthenticationException(e.getMessage());
        }
        
    }

    private void validate()
    {
        if(connection ==null)
        {
            logger.error("Connection is not initialized!");
            throw new AuthenticationException("Connection is not initialized!"+connection);
        }
    }

    @Override
    public int getResponseCode()
    {
        validate();
        try
        {
            return connection.getResponseCode();
        }
        catch (IOException e)
        {
            logger.error(e.getMessage());
            throw new AuthenticationException(e.getMessage());
        }
    }

    @Override
    public void collectTweets(String tweet, String hbaseMasterNode,String hbaseMasterPort, String zookeeperHost, String trendName)
    {
    	System.out.println("Inside collect tweetes" + trendName);
       messageQueue.add(tweet);
      (new Thread(new Dequeuer(messageQueue,hbaseMasterNode,hbaseMasterPort,zookeeperHost, trendName,service))).start();
        
    }

}
